package com.example.demo.util;

import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.Result;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.util.EnvUtil;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.repository.RepositoryDirectoryInterface;
import org.pentaho.di.repository.kdr.KettleDatabaseRepository;
import org.pentaho.di.repository.kdr.KettleDatabaseRepositoryMeta;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;

/**
 * 执行本地kettle文件(ktr,kjb)
 *
 */
public class KettleUtil {
    /**
     * 本地kettle文件转换
     * @param params
     * @param path
     * @return
     * @throws Exception
     */
    public static Result runKtr(String[] params, String path) throws Exception {

        Trans trans = null;

        KettleEnvironment.init();// 初始化
        EnvUtil.environmentInit();

        TransMeta transMeta = new TransMeta(path);
        // 转换
        trans = new Trans(transMeta);
        trans.execute(params);

        // 等待转换执行结束
        trans.waitUntilFinished();

        // 抛出异常
        if (trans.getErrors() > 0) {
            throw new KettleException("There are errors during transformation exception!(传输过程中发生异常)");
        }

        return trans.getResult();
    }

    /**
     * 本地job工作
     * @param params
     * @param path
     * @return
     * @throws Exception
     */
    public static Result runJob(String[] params, String path)  throws Exception {

        KettleEnvironment.init();

        JobMeta jobMeta = new JobMeta(path, null);
        Job job = new Job(null, jobMeta);

        // 向Job 脚本传递参数，脚本中获取参数值：${参数名}
        // job.setVariable("id", params[0]);
        // job.setVariable("content", params[1]);
        // job.setVariable("file", params[2]);

        job.start();

        job.waitUntilFinished();

        if (job.getErrors() > 0) {
            throw new KettleException("There are errors during job exception!(执行job发生异常)");
        }

        return job.getResult();
    }

    /**
     * 执行资源库的中的转换。
     */
    public static void runWithDb() throws KettleException{
        KettleEnvironment.init();
        //创建DB资源库
        KettleDatabaseRepository repository=new KettleDatabaseRepository();
        DatabaseMeta databaseMeta=new DatabaseMeta("kettle","mysql","jdbc","localhost","kettle","3306","root","root");
        //选择资源库
        KettleDatabaseRepositoryMeta kettleDatabaseRepositoryMeta=new KettleDatabaseRepositoryMeta("kettle","kettle","Transformation description",databaseMeta);
        repository.init(kettleDatabaseRepositoryMeta);
        //连接资源库
        repository.connect("admin","admin");
        RepositoryDirectoryInterface directoryInterface=repository.loadRepositoryDirectoryTree();
        //选择转换
        TransMeta transMeta=repository.loadTransformation("demo1",directoryInterface,null,true,null);
        Trans trans=new Trans(transMeta);
        trans.execute(null);
        trans.waitUntilFinished();//等待直到数据结束
        if(trans.getErrors()>0){
            System.out.println("transformation error");
        }else{
            System.out.println("transformation successfully");
        }
    }

}
